package com.liux.rabbitmq.four;

import com.liux.rabbitmq.utils.RabbitMQUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @Date 2022/4/25 10:03
 * @Description
 *  发布确认模式
 *      1、单个确认
 *      2、批量确认
 *      3、异步批量确认
 **/
public class ConfirmMessage {
    //批量发消息的个数
    public static final int MESSAGE_COUNT =1000;
    public static void main(String[] args) {
        try {
            //单个确认
            //publishMessageIndividually(); //单个确认模式，发布1000个消息，用时758毫秒
            //批量确认
            //publishMessageBatch(); //批量确认模式，发布1000个消息，用时136毫秒
            //异步确认模式
            publishMessageAsync();//异步确认模式，发布1000个消息，用时72毫秒


            /**
             * 对concurrentSkipListMap.headMap删除的实验，结果没问题
            ConcurrentSkipListMap<Long,String> concurrentSkipListMap =new ConcurrentSkipListMap<>();
            concurrentSkipListMap.put(1L, "1");
            concurrentSkipListMap.put(2L, "2");
            concurrentSkipListMap.put(3L, "3");
            concurrentSkipListMap.put(4L, "4");
            concurrentSkipListMap.put(5L, "5");
            concurrentSkipListMap.put(6L, "6");
            concurrentSkipListMap.put(7L, "7");
            for(Map.Entry<Long,String> entry: concurrentSkipListMap.entrySet()){
                System.out.println(entry.getValue());
            }
             //获取headMap指定Key之前的值，true是<= false是<
            ConcurrentNavigableMap<Long, String> deleteMap = concurrentSkipListMap.headMap(6L,true);
            deleteMap.clear();
            System.out.println("==============================================================");
            for(Map.Entry<Long,String> entry: concurrentSkipListMap.entrySet()){
                System.out.println(entry.getValue());
            }
             **/
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /**
     * 单个确认模式
     * @throws Exception
     */
    public  static  void  publishMessageIndividually() throws Exception {
         Channel channel = RabbitMQUtil.getChannel();
         //队列名称生成
        String queueName = UUID.randomUUID().toString();
        //队列声明
        channel.queueDeclare(queueName,false,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //开始时间
        long begin = System.currentTimeMillis();
        //批量发消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message=i+"";
            channel.basicPublish("",queueName,null,message.getBytes(StandardCharsets.UTF_8));
            //单个确认
            boolean flag = channel.waitForConfirms();
        }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("单个确认模式，发布"+MESSAGE_COUNT+"个消息，用时"+(end-begin)+"毫秒");

    }

    /**
     * 批量确认模式
     * @throws Exception
     */
    public  static  void  publishMessageBatch() throws Exception {
        Channel channel = RabbitMQUtil.getChannel();
        //队列名称生成
        String queueName = UUID.randomUUID().toString();
        //队列声明
        channel.queueDeclare(queueName,false,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //开始时间
        long begin = System.currentTimeMillis();
        //设置每多少条，批量确认
        int batchSize=100;
        //批量发消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message=i+"";
            channel.basicPublish("",queueName,null,message.getBytes(StandardCharsets.UTF_8));
            //批量确认,每100条确认一次
            if((i+1)%batchSize==0){
                channel.waitForConfirms();
            }
        }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("批量确认模式，发布"+MESSAGE_COUNT+"个消息，用时"+(end-begin)+"毫秒");

    }

    /**
     * 异步确认模式
     * @throws Exception
     */
    public  static  void  publishMessageAsync() throws Exception {
        Channel channel = RabbitMQUtil.getChannel();
        //队列名称生成
        String queueName = UUID.randomUUID().toString();
        //队列声明
        channel.queueDeclare(queueName,false,false,false,null);
        //开启发布确认
        channel.confirmSelect();

        /**
         * 需要一个线程安全、有序的一个哈希表，适用于高并发
         * 1、轻松的将序号和消息进行绑定
         * 2、轻松的批量删除，只需给序号
         * 3、支持高并发(多线程)
         */
        ConcurrentSkipListMap<Long,String> concurrentSkipListMap =new ConcurrentSkipListMap<>();
        //开始时间
        long begin = System.currentTimeMillis();

        //发送成功回调接口
        /**
         * var1:消息的标机
         * var3:是否批量确认
         */
        ConfirmCallback ackConfirmCallback =(var1,var3) ->{
            //2、删除已经确认的消息，剩下的就是未确认的消息
            System.out.println("接收到的序号为："+var1);
            if(var3){
                System.out.println("批量处理模式");
                //如果是批量确认,采用这种方式能批量删除消息，在main里面有实验
                ConcurrentNavigableMap<Long, String> confirmMap = concurrentSkipListMap.headMap(var1,true);
                System.out.println("Map长度为"+ concurrentSkipListMap.size());
                confirmMap.clear();
                System.out.println("Map长度为"+concurrentSkipListMap.size());
            }else {
                System.out.println("单个处理模式");
                //如果是单个确认，则一个个删除
                concurrentSkipListMap.remove(var1);
            }
            for(Map.Entry<Long,String> entry: concurrentSkipListMap.entrySet()){
                System.out.println(entry.getValue());
            }
        };
        //发送失败回调接口
        ConfirmCallback nackConfirmCallback =(var1,var3) ->{
            //3、打印未确认的消息
            System.out.println("消息失败："+var1);
        };
        //准备消息的监听器，监听消息哪些成功了，哪一些失败了
        channel.addConfirmListener(ackConfirmCallback,nackConfirmCallback);
        //批量发消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message=i+"";
            channel.basicPublish("",queueName,null,message.getBytes(StandardCharsets.UTF_8));
            //1、记录发送的消息
            concurrentSkipListMap.put(channel.getNextPublishSeqNo()-1,message);
        }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("异步确认模式，发布"+MESSAGE_COUNT+"个消息，用时"+(end-begin)+"毫秒");

    }
}
